package com.zytech.zspider.eventbus.rabbitmq;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
import com.zytech.modules.lang.Langs;
import com.zytech.modules.lang.Requires;
import com.zytech.zspider.eventbus.EventEnvelope;
import com.zytech.zspider.eventbus.support.EventOperations;

import java.io.ByteArrayOutputStream;
import java.io.IOException;

/**
 * @author siuming
 */
final class RabbitMQOperations implements EventOperations {

    private final ConnectionFactory connectionFactory;

    /**
     * @param connectionFactory
     */
    public RabbitMQOperations(ConnectionFactory connectionFactory) {
        Requires.notNull(connectionFactory);
        this.connectionFactory = connectionFactory;
    }

    @Override
    public final void sendToQueue(String qname, EventEnvelope envelope) {
        try {

            RabbitMQChannels
                    .ofQueueChannel(getConnectionFactory(), qname)
                    .basicPublish("", qname, MessageProperties.PERSISTENT_BASIC, ofMessageBodyBy(envelope));
        } catch (IOException e) {
            throw Langs.toUncheck(e);
        }
    }

    /**
     * @param eventEnvelope
     * @return
     */
    protected byte[] ofMessageBodyBy(EventEnvelope eventEnvelope) {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        Langs.dumps(eventEnvelope, out);
        return out.toByteArray();
    }

    @Override
    public final void sendToTopic(String tname, EventEnvelope envelope) {
        try {

            RabbitMQChannels
                    .ofTopicChannel(getConnectionFactory(), tname)
                    .basicPublish(tname, "", MessageProperties.PERSISTENT_BASIC, ofMessageBodyBy(envelope));
        } catch (IOException e) {
            throw Langs.toUncheck(e);
        }
    }

    /**
     * @return
     */
    protected final ConnectionFactory getConnectionFactory() {
        return connectionFactory;
    }
}
